package com.gzc.just.play.last.war.common.handler;

import com.gzc.just.play.last.war.common.serializer.AutoMessageDeserializerFactory;
import com.gzc.just.play.last.war.common.serializer.MessageDeserializer;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 客户端会话类，封装一个客户端连接的消息处理逻辑
 * 每个客户端会话对应一个虚拟线程
 */
public class ClientSession {
    private static final Logger logger = LoggerFactory.getLogger(ClientSession.class);
    
    private final long sessionId;
    private final ChannelHandlerContext ctx;
    private final VirtualThreadPerClientHandler handler;
    private final Thread processingThread;
    private final AtomicBoolean running;
    
    // 消息队列，线程安全
    private final ConcurrentLinkedQueue<MessageContext> messageQueue = new ConcurrentLinkedQueue<>();
    
    // 消息处理器映射
    private final ConcurrentHashMap<Integer, SimpleMessageHandler> messageHandlers;
    
    /**
     * 消息上下文，封装消息和相关信息
     */
    public static class MessageContext {
        private final int messageId;
        private final byte[] messageBytes;
        private final long timestamp;
        
        public MessageContext(int messageId, byte[] messageBytes) {
            this.messageId = messageId;
            this.messageBytes = messageBytes;
            this.timestamp = System.currentTimeMillis();
        }
        
        public int getMessageId() {
            return messageId;
        }
        
        public byte[] getMessageBytes() {
            return messageBytes;
        }
        
        public long getTimestamp() {
            return timestamp;
        }
    }
    
    /**
     * 构造函数
     * @param sessionId 会话ID
     * @param ctx Netty通道上下文
     * @param handler 消息处理器
     * @param messageHandlers 消息处理器映射
     */
    public ClientSession(long sessionId, ChannelHandlerContext ctx, 
                        VirtualThreadPerClientHandler handler,
                        ConcurrentHashMap<Integer, SimpleMessageHandler> messageHandlers) {
        this.sessionId = sessionId;
        this.ctx = ctx;
        this.handler = handler;
        this.messageHandlers = messageHandlers;
        this.running = new AtomicBoolean(false);
        
        // 创建并启动专用的虚拟线程
        this.processingThread = Thread.ofVirtual()
                .name("ClientSession-" + sessionId)
                .start(this::processMessagesLoop);
        
        // 设置为守护线程
        this.processingThread.setDaemon(true);
    }
    
    /**
     * 启动会话处理
     */
    public void start() {
        running.set(true);
        logger.info("Started client session: {} in thread: {}", sessionId, Thread.currentThread().getName());
    }
    
    /**
     * 停止会话处理
     */
    public void stop() {
        running.set(false);
        
        // 等待处理线程结束
        if (processingThread != null && processingThread.isAlive()) {
            try {
                processingThread.join(2000); // 最多等待2秒
            } catch (InterruptedException e) {
                logger.warn("Interrupted while waiting for session thread to stop: {}", sessionId, e);
                Thread.currentThread().interrupt();
            }
        }
        
        logger.info("Stopped client session: {}", sessionId);
    }
    
    /**
     * 提交消息到处理队列
     * @param messageId 消息ID
     * @param messageBytes 消息字节数组
     * @return 是否成功提交
     */
    public boolean submitMessage(int messageId, byte[] messageBytes) {
        if (!running.get()) {
            logger.warn("Session {} is not running, ignoring message: {}", sessionId, messageId);
            return false;
        }
        
        // 查找消息处理器
        SimpleMessageHandler handler = messageHandlers.get(messageId);
        if (handler == null) {
            logger.warn("No handler found for message ID: {} in session: {}", messageId, sessionId);
            return false;
        }
        
        // 将消息加入队列
        MessageContext messageContext = new MessageContext(messageId, messageBytes);
        messageQueue.offer(messageContext);
        
        logger.debug("Submitted message {} to session {} queue size: {}", 
                    messageId, sessionId, messageQueue.size());
        
        return true;
    }
    
    /**
     * 处理消息的主循环
     * 在专用的虚拟线程中运行
     */
    private void processMessagesLoop() {
        logger.info("Started message processing loop for session: {} in thread: {}", 
                   sessionId, Thread.currentThread().getName());
        
        try {
            while (running.get() && !Thread.currentThread().isInterrupted()) {
                try {
                    // 从队列中取出消息
                    MessageContext messageContext = messageQueue.poll();
                    if (messageContext == null) {
                        // 没有消息时短暂休眠，避免空转
                        Thread.sleep(1);
                        continue;
                    }
                    
                    // 处理消息
                    processMessage(messageContext);
                    
                } catch (InterruptedException e) {
                    logger.info("Session {} processing thread interrupted", sessionId);
                    break;
                } catch (Exception e) {
                    logger.error("Error in session {} processing loop", sessionId, e);
                }
            }
        } finally {
            // 处理剩余消息
            while (!messageQueue.isEmpty()) {
                MessageContext messageContext = messageQueue.poll();
                if (messageContext != null) {
                    try {
                        processMessage(messageContext);
                    } catch (Exception e) {
                        logger.error("Error processing remaining message in session: {}", sessionId, e);
                    }
                }
            }
            
            logger.info("Stopped message processing loop for session: {}", sessionId);
        }
    }
    
    /**
     * 处理单个消息
     * 使用统一的反序列化器注册表获取反序列化器
     * @param messageContext 消息上下文
     */
    private void processMessage(MessageContext messageContext) {
        int messageId = messageContext.getMessageId();
        byte[] messageBytes = messageContext.getMessageBytes();
        
        try {
            // 查找消息处理器
            SimpleMessageHandler messageHandler = messageHandlers.get(messageId);
            if (messageHandler == null) {
                logger.warn("No handler found for message ID: {} in session: {}", messageId, sessionId);
                return;
            }
            
            // 从统一注册表获取反序列化器
            MessageDeserializer deserializer = AutoMessageDeserializerFactory.getDeserializerStatic(messageId);
            if (deserializer == null) {
                logger.error("No deserializer registered for message ID: {} in session: {}. " +
                           "Please create a @MessageDeserializer annotated class for this message.", 
                           messageId, sessionId);
                return;
            }
            
            // 反序列化消息
            Object message = deserializer.deserialize(messageBytes);
            if (message == null) {
                logger.error("Failed to deserialize message with ID: {} in session: {}", messageId, sessionId);
                return;
            }
            
            // 处理消息
            messageHandler.handle(message, ctx);
            
        } catch (Exception e) {
            logger.error("Error processing message {} in session: {}", messageId, sessionId, e);
        }
    }
    
    /**
     * 获取会话ID
     */
    public long getSessionId() {
        return sessionId;
    }
    
    /**
     * 获取通道上下文
     */
    public ChannelHandlerContext getChannelContext() {
        return ctx;
    }
    
    /**
     * 获取当前队列中的消息数量
     */
    public int getQueueSize() {
        return messageQueue.size();
    }
    
    /**
     * 检查会话是否正在运行
     */
    public boolean isRunning() {
        return running.get();
    }
}